
package hasthi.coordinator;

import static hasthi.common.constants.HasthiConstants.LOGGER_NAME;
import hasthi.actuators.ActionContext;
import hasthi.actuators.event.HasthiEvent;
import hasthi.common.BaseContext;
import hasthi.common.DistributedLogger;
import hasthi.common.ErrorManager;
import hasthi.common.HasthiException;
import hasthi.common.InvaliedResourceDataException;
import hasthi.common.RemoteProcessorException;
import hasthi.common.Utils;
import hasthi.common.XmlBeanUtils;
import hasthi.common.HasthiException.FaultCode;
import hasthi.common.constants.HasthiConstants;
import hasthi.common.constants.WSDMConstants;
import hasthi.common.constants.WSDMConstants.SystemState;
import hasthi.container.MessageContext;
import hasthi.container.SimpleMessageContext;
import hasthi.container.UnicastContianier;
import hasthi.decision.RuleEngine;
import hasthi.manager.AssignResourcesDocument;
import hasthi.manager.AssignResourcesResponseDocument;
import hasthi.manager.CoordinatorStateRequestDocument;
import hasthi.manager.CoordinatorStateResponeDocument;
import hasthi.manager.ExecuteQueryDocument;
import hasthi.manager.ExecuteQueryResponseDocument;
import hasthi.manager.GetResourceHoldersDocument;
import hasthi.manager.GetResourceHoldersResponseDocument;
import hasthi.manager.JoinManagementRingDocument;
import hasthi.manager.JoinManagementRingResponseDocument;
import hasthi.manager.LeaveManagementRingDocument;
import hasthi.manager.LeaveManagementRingResponseDocument;
import hasthi.manager.ListAllResourcesDocument;
import hasthi.manager.ListAllResourcesResponseDocument;
import hasthi.manager.ManageMeRequestDocument;
import hasthi.manager.ManagedResourceType;
import hasthi.manager.ManagerAddress;
import hasthi.manager.ManagerContext;
import hasthi.manager.ManagerHeartBeatResponseDocument;
import hasthi.manager.ManagerHeartbeatDocument;
import hasthi.manager.NotifyEventDocument;
import hasthi.manager.QueryResults;
import hasthi.manager.ResolveServiceDependancyByTypeDocument;
import hasthi.manager.ResolveServiceDependancyByTypeResponseDocument;
import hasthi.manager.SuspectFaultyDocument;
import hasthi.manager.UnAssignResourceDocument;
import hasthi.manager.UserInterventionDocument;
import hasthi.manager.CoordinatorStateResponeDocument.CoordinatorStateRespone;
import hasthi.manager.ExecuteQueryDocument.ExecuteQuery;
import hasthi.manager.ExecuteQueryDocument.ExecuteQuery.Query;
import hasthi.manager.ExecuteQueryResponseDocument.ExecuteQueryResponse;
import hasthi.manager.GetResourceHoldersResponseDocument.GetResourceHoldersResponse;
import hasthi.manager.JoinManagementRingResponseDocument.JoinManagementRingResponse;
import hasthi.manager.ListAllResourcesResponseDocument.ListAllResourcesResponse;
import hasthi.manager.ManageMeRequestDocument.ManageMeRequest;
import hasthi.manager.ManageMeRequestDocument.ManageMeRequest.ResourceDetails;
import hasthi.manager.ManagerHeartBeatResponseDocument.ManagerHeartBeatResponse;
import hasthi.manager.NotifyEventDocument.NotifyEvent;
import hasthi.manager.ResolveServiceDependancyByTypeResponseDocument.ResolveServiceDependancyByTypeResponse;
import hasthi.manager.ResolveServiceDependancyByTypeResponseDocument.ResolveServiceDependancyByTypeResponse.ServiceData;
import hasthi.manager.UserInterventionDocument.UserIntervention;
import hasthi.manager.UserInterventionDocument.UserIntervention.ServiceStatusChange;
import hasthi.manager.monitering.EventListener;
import hasthi.manager.monitering.MoniteringAgent;
import hasthi.metamodel.SummerizedResourceModel;
import hasthi.metamodel.resource.EditableManagedResource;
import hasthi.metamodel.resource.ManagedResource;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;

import javax.xml.namespace.QName;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.xmlbeans.XmlObject;

/**
 * This code has the logic for Coordinator message processing
 * @author Srinath Perera(hperera@cs.indiana.edu)
 */

public class CoordinatorMessageProcessor {
    private Log log = LogFactory.getLog(LOGGER_NAME);
    private CoordinatorContext coordinatorContext;
    private ManagerContext managerContext;
    private MoniteringAgent moniteringAgent;
    private Map<QName,QName> resourcesBeingAdded = new Hashtable<QName, QName>();
    private DistributedLogger dlogger;
     
    
    public CoordinatorMessageProcessor(CoordinatorContext coordinatorContext,
            ManagerContext managerContext) throws HasthiException {
        this.coordinatorContext = coordinatorContext;
        this.managerContext = managerContext;
        
        BaseContext baseConfigContext = managerContext.getBaseContext();
        //subscribe to Notification channel 
        
        if(!baseConfigContext.isJoinViaBootstrap()){
            moniteringAgent = new MoniteringAgent(baseConfigContext.getMessageBrokerUrl(),baseConfigContext.getAdminTopic());
            moniteringAgent.addTrigger(new EventListener() {
                public void performTriggerAction(XmlObject event, XmlObject[] matchingResults)
                        throws HasthiException {
                    log.info("Event recived "+ event);
                    if(event instanceof ManageMeRequestDocument){
                        processManageMeMessage((ManageMeRequestDocument)event);    
                    }
                }
                public String getSearchPath() {
                    return null;
                }
            });
            moniteringAgent.startMonitering();
        }
        dlogger = managerContext.getBaseContext().getDlogger();
        //start the coordinator demon
        coordinatorContext.getCoordinationDemon().startDeamon();
        //Timer timer = new Timer(true);
        //timer.scheduleAtFixedRate(coordinatorContext.getCoordinationDemon(), delay, period);
    }

    /**
     * Add the given manager to managment ring and return sucessor array and rank assigned to new Manager
     * @param document
     * @return
     */
    public JoinManagementRingResponseDocument joinManagmentRing(JoinManagementRingDocument document){
        JoinManagementRingResponseDocument  responseDocument = JoinManagementRingResponseDocument.Factory.newInstance();
        JoinManagementRingResponse response = responseDocument.addNewJoinManagementRingResponse();
        
        ManagerAddress address = document.getJoinManagementRing().getAddress();
        int assignedRank = coordinatorContext.addNewManager(address);
        response.setAssignedRank(assignedRank);
        coordinatorContext.copySucessors(response);
        response.setManagerCount(coordinatorContext.getManagerCount());
        response.setCoordinatorAddress(managerContext.getMyAddress());
        response.setCoordinatorTimestamp(managerContext.getCoordinatorTimeStamp());
        return responseDocument;
    }
    
    /**
     * Leave the managerfrom managment ring
     * @param document
     * @return
     */
    public LeaveManagementRingResponseDocument leaveManagmentRing(LeaveManagementRingDocument document){
        LeaveManagementRingResponseDocument  responseDocument = LeaveManagementRingResponseDocument.Factory.newInstance();
        responseDocument.addNewLeaveManagementRingResponse();
        
        ManagerAddress address = document.getLeaveManagementRing().getAddress();
        ResourceAssignmentUtils.removeAndReassignResources(coordinatorContext,address);
        dlogger.info(address.getSoapAddress() + " left managment ring");
        return responseDocument;
    }
    
    /**
     * Given a resource name, return managers who holds the resource
     * @param document
     * @return
     */
    public GetResourceHoldersResponseDocument getResourceHolders(GetResourceHoldersDocument document){
        GetResourceHoldersResponseDocument responseDocument =  GetResourceHoldersResponseDocument.Factory.newInstance();
        GetResourceHoldersResponse response = responseDocument.addNewGetResourceHoldersResponse();
        
        QName[] resourceNames = document.getGetResourceHolders().getResourceNameArray();
        List<ManagerAddress> results = new ArrayList<ManagerAddress>(resourceNames.length);
        for(QName resourceName:resourceNames){
            RemoteManagerContext remoteManager = coordinatorContext.getManagerForResource(resourceName); 
            if(remoteManager != null){
                results.add(remoteManager.getAddress());    
            }
        }
        if(results.size() > 0){
            response.setResourceHolderArray(results.toArray(new ManagerAddress[0]));    
        }
        return responseDocument;
    }
    
    
    /**
     * When a Management heartbeat is received, update the local the object based on updated parameters. This message can
     * could effectively act as the Join message for the Management ring, and response is sent using piggyback parameters.
     * However if some resource is already assigned 
     * to someone else,  Manager will be asked to stop listening to the resource using piggyback parameters.  
     * @param document
     * @return
     * @throws HasthiException
     */
    
    public ManagerHeartBeatResponseDocument processManagerHeartBeat(ManagerHeartbeatDocument document) throws HasthiException{
        try {
            ManagerHeartBeatResponseDocument responseDocument = ManagerHeartBeatResponseDocument.Factory.newInstance();
            ManagerHeartBeatResponse heartBeatResponse = responseDocument.addNewManagerHeartBeatResponse();
            heartBeatResponse.setCoordinatorTimestamp(managerContext.getCoordinatorTimeStamp());
            
            String managerID = Utils.mngAdr2Id(document.getManagerHeartbeat().getAddress());
            RemoteManagerContext managerData = coordinatorContext.getManagerData(managerID);
            
            boolean isNewResources = document.getManagerHeartbeat().getAllResourceIncluded();
            if(managerData == null){
                /**
                 * This could be a Manager who is Joining after a coordination failure. Add it as a new manager.
                 */
                int assignedRank = coordinatorContext.addNewManager(document.getManagerHeartbeat().getAddress());
                heartBeatResponse.setAddedAsNew(true);
                heartBeatResponse.setAssignedRank(assignedRank);
                managerData = coordinatorContext.getManagerData(Utils.mngAdr2Id(document.getManagerHeartbeat().getAddress()));
            }else if(isNewResources 
                && !Utils.equals(document.getManagerHeartbeat().getAddress(), managerContext.getMyAddress())){
                log.info(Utils.getID(document.getManagerHeartbeat().getAddress()) + " Send all resources, but not added as new");
            }
            long currentTime = System.currentTimeMillis();
            managerData.setLastUpdated(currentTime);
//            /**
//             * Mark all resources as updated
//             */
//            List<QName> resourcesOfThisManager = managerData.getAssignedResources();
            SummerizedResourceModel resourceModel = coordinatorContext.getResourceModel();
//            int resourceCount4thisManager = resourcesOfThisManager.size();
//            for(int i =0;i<resourceCount4thisManager;i++){
//                QName resourceName = resourcesOfThisManager.get(i);
//                EditableManagedResource resource = resourceModel.getResource(resourceName);
//                
//                //This makes sure handled resources will be removed from the system eventually becouse their 
//                //last update time would not change
//                if(!resource.getState().equals(SystemState.CrashedState.toString())
//                        && !resource.getState().equals(SystemState.StoppedState.toString())
//                        && !resource.getState().equals(SystemState.RepairedState.toString())){
//                    resource.setLastUpdated(currentTime);    
//                }
//                
//            }
            
            ManagedResourceType[] resourceUpdates = document.getManagerHeartbeat().getPiggybackUpdateArray();
            RuleEngine engine = coordinatorContext.getRuleEngine();
            for(ManagedResourceType resourceUpdate:resourceUpdates){
                EditableManagedResource resource  = resourceModel.getResource(resourceUpdate.getName());
                if(resource == null){
                    //if this is not a new manager, resource it send might be already removed from the
                    //coordinator. So ask manager to remove it
                    boolean iSresourceBeingAdded = resourcesBeingAdded.containsKey(resourceUpdate.getName());
                    if(!iSresourceBeingAdded && isNewResources){
                        //New resource, not assigned to anyone. So we let the manager keep his assignment
                        resource =  resourceModel.addNewResource(resourceUpdate.getName(), resourceUpdate);
                        coordinatorContext.addResource2Manager(resource.getQName(), managerData);
                    }else{
                        //This resource is being assigned to a manager
                        heartBeatResponse.addNewResourceNameToRemove().setQNameValue(resourceUpdate.getName());
                        log.warn("Asking Manager "+Utils.getID(managerData.getAddress()) 
                                + " to stop managing the resource "+ resourceUpdate.getName() 
                                + (iSresourceBeingAdded?" Becouse it is being added":""));
                    }
                }else{
                    //We need to check does this resource is already assigned to someone else
                    RemoteManagerContext managerForResource = coordinatorContext.getManagerForResource(resourceUpdate.getName());
                    if(managerForResource == null){
                        log.warn("Resource update "+ resourceUpdate.getName() + " received before assigned to a manager. ");
                    }else if(Utils.equals(managerForResource.getAddress(), managerData.getAddress())){
                        boolean changed = XmlBeanUtils.processProperties(resourceUpdate,new ResourceUpdater(resource));
                        //boolean changed = false;
                        //Map<QName,Object> properties  = XmlBeanUtils.extractProperties(resourceUpdate);
                        //for(QName name:properties.keySet()){
                        //    resource.setProperty(name.getLocalPart(), properties.get(name));
                        //    changed = true;
                        //}
                        //Signal to rule engine that the resource has changed
                        if(changed){
                            engine.objectChanged(resource);    
                        }
                    }else{
                        //We instruct the manager to stop listening to the resource
                        heartBeatResponse.addNewResourceNameToRemove().setQNameValue(resourceUpdate.getName());
                    }
                }
                
            }
            if(isNewResources){
                log.info("["+managerContext + "]Add " + document.getManagerHeartbeat().getPiggybackUpdateArray().length +"/"
                        + coordinatorContext.getResourceCount()+" Resources from  " 
                        + Utils.getID(managerData.getAddress()) +" "+ coordinatorContext.getManagerCount() + "th Manager");
            }
            
            coordinatorContext.copySucessors(heartBeatResponse);
            heartBeatResponse.setManagerCount(coordinatorContext.getManagerCount());
            return responseDocument;
        } catch (InvaliedResourceDataException e) {
            throw new HasthiException(e,FaultCode.InvaliedRequest);
        }
    }
    
    public void processManageMeMessage(ManageMeRequestDocument requestDocument){
        ManageMeRequest manageMeRequest = requestDocument.getManageMeRequest();
        
        ResourceDetails[] resourceDetails = manageMeRequest.getResourceDetailsArray();
        
        for(ResourceDetails resourceDetail:resourceDetails){
            QName resourceName = resourceDetail.getResourceName();
            ManagerAddress address = null;
            try {
                dlogger.info("Join Request(" +resourceName + ")");
                //If we have already assigned this resource, ignore
                if(resourcesBeingAdded.containsKey(resourceName)){
                    log.warn("Manage Me message from "+ resourceName +"/"
                            +resourceDetail.getManagmentAgentEpr() + " ignored as it is already being added");
                    return;
                } else{
                    EditableManagedResource resource = coordinatorContext.getResourceModel().getResource(resourceName);
                    if(resource != null){
                        if(Utils.isActiveResource(resource)){
                            log.warn("Manage Me message from "+ resourceName +"/"
                                    +resourceDetail.getManagmentAgentEpr() + " ignored as it is already assigned");
                            return;
                        }else{
                            //if a failed resource trying to rejoin, we remove that resource and let new one join
                            log.info("faulty resource " +resourceName + " removed becouse a resource with the same name joined");
                            coordinatorContext.getResourceModel().removeResource(resourceName);
                            coordinatorContext.removeResource(resourceName);  
                            coordinatorContext.getRuleEngine().removeObject(resource);
                        }
                    }  
                } 
                resourcesBeingAdded.put(resourceName, resourceName);
                coordinatorContext.getManageMeMessageBag().addManagemeRequest(resourceName, resourceDetail.getManagmentAgentEpr());
            }catch(RemoteProcessorException e){
                if(FaultCode.ErrorAtDependentService.equals(e.getFaultCode())){
                    dlogger.errror("Resource Management Endpoint is Down "+ resourceName.getLocalPart() , e);
                }else{
                    dlogger.errror("Error processing ManageMeRequest for "+ resourceName.getLocalPart() , e);    
                }
            } catch (HasthiException e) {
                dlogger.errror("Error processing ManageMeRequest for "+ resourceName.getLocalPart() , e);
            }
                //Find the best Manager to assign the resource
                
//                RemoteManagerContext resourceCaretaker = coordinatorContext.findBestManager4Resource(resourceName);
//                address = resourceCaretaker.getAddress();
//                
//                //Assign the resource to the selected Manager
//                UnicastContianier soapContainer = managerContext.getUnicastContainer(); 
//                String managementEprOfResource = requestDocument.getManageMeRequest().getManagmentAgentEpr();
//                
//                AssignResourceDocument assignResourceDocument = ManagerClientUtils.createAssignResourceRequest(managementEprOfResource, resourceName);
//                MessageContext requestMessage  = new SimpleMessageContext(assignResourceDocument,address.getSoapAddress(),
//                        ASSIGN_RESOURCE_ACTION); 
//                AssignResourceResponseDocument response = (AssignResourceResponseDocument)soapContainer.sendReceive(requestMessage);
            
        }
        
  
    }
    
    public void processAssignResourceResponse(AssignResourcesResponseDocument response) throws HasthiException, InvaliedResourceDataException{
        ManagedResourceType[] resourceSummerylist = response.getAssignResourcesResponse().getResourceSummeryArray();
        String managerID = Utils.mngAdr2Id(response.getAssignResourcesResponse().getManagerAddress());
        RemoteManagerContext resourceCaretaker = coordinatorContext.getManagerData(managerID);
        for(ManagedResourceType resourceSummery:resourceSummerylist){
            //build the resource model using summery about the resource returned by the Assignee
            SummerizedResourceModel resourceModel = coordinatorContext.getResourceModel();
            QName name = resourceSummery.getName();
            //TODO if management endpoint is null, that means contacting resource has failed. We will remove the resource
            //from the resources being added and continue.
            if(resourceSummery.getManagementEndpoint() != null && resourceSummery.getManagementEndpoint().trim().length() > 0){
                EditableManagedResource resource = resourceModel.addNewResource(name, resourceSummery);
                
                coordinatorContext.getRuleEngine().addObject(resource);
                
                //Assign resource the the manaeger, we do this late as possible
                coordinatorContext.addResource2Manager(name, resourceCaretaker);
                dlogger.debug("Join Completed ("+ name + " to "+ managerID + ")");
            }
            resourcesBeingAdded.remove(name);
        }
    }
    
    
    public ListAllResourcesResponseDocument listAllResources(ListAllResourcesDocument document) throws HasthiException{
        Collection<QName> allresources = coordinatorContext.getResourceModel().getResources();
        ListAllResourcesResponseDocument responseDocument = ListAllResourcesResponseDocument.Factory.newInstance();
        ListAllResourcesResponse response = responseDocument.addNewListAllResourcesResponse();
        if(allresources != null){
            ArrayList<ManagedResourceType> resources = new ArrayList<ManagedResourceType>(allresources.size());
            for(QName name:allresources){
                ManagedResource resource = coordinatorContext.getResourceModel().getResource(name); 
                resources.add(resource.toXml(true));
            }
            response.setResourceDataArray(resources.toArray(new ManagedResourceType[0]));
        }
        return responseDocument;
    }
    
    public CoordinatorStateResponeDocument getCoordinatorState(CoordinatorStateRequestDocument requestDocument) throws HasthiException{
        CoordinatorStateResponeDocument responeDocument = CoordinatorStateResponeDocument.Factory.newInstance();
        CoordinatorStateRespone response = responeDocument.addNewCoordinatorStateRespone();
        response.setManagementRing(coordinatorContext.toXml());
        return responeDocument;
    }
    
    public class AsyncCoordinatorProcessor implements Runnable{
        private MessageContext messageContext;
        public AsyncCoordinatorProcessor(MessageContext messageContext) {
             
            this.messageContext = messageContext;
            
        }
        public void run(){
            XmlObject request= null;
            try {
                request = messageContext.getRequest();
                if(request instanceof ManageMeRequestDocument){
                    processManageMeMessage((ManageMeRequestDocument)request);
                }else if(request instanceof AssignResourcesResponseDocument){
                    processAssignResourceResponse((AssignResourcesResponseDocument)request);       
                }else{
                    log.error("Unknown message at async processing of coordinator "+XmlBeanUtils.prettyPrint(request));
                }
            } catch (Throwable e) {
                log.error("Error while processesing "+XmlBeanUtils.prettyPrint(request), e);
                ErrorManager.error(e);
            }
        }
    }
    
    public AsyncCoordinatorProcessor createAsyncCoordinatorProcessor(MessageContext messageContext){
        return new AsyncCoordinatorProcessor(messageContext);
    }
    
    public void notifyEvent(NotifyEventDocument eventDocument){
        ActionContext actionContext = coordinatorContext.getRuleEngine().getActionCenter().getActionContext();
        NotifyEvent notifyEvent = eventDocument.getNotifyEvent();
        actionContext.getEventCorrelator().triggerEvent(
                new HasthiEvent(notifyEvent.getName(),notifyEvent.getEventDescription(),notifyEvent.getContent()));
    }
    
    public ResolveServiceDependancyByTypeResponseDocument resolveServiceDependancy(ResolveServiceDependancyByTypeDocument document){
        List<ManagedResource> managedResources;
        
        //there is a remote chance that rule engine throw a error while doing a query, so we retry
        ResolveServiceDependancyByTypeResponseDocument responseDocument = ResolveServiceDependancyByTypeResponseDocument.Factory.newInstance();
        ResolveServiceDependancyByTypeResponse resolveServiceDependancyByTypeResponse = responseDocument.addNewResolveServiceDependancyByTypeResponse();

        QName[] portTypeNames = document.getResolveServiceDependancyByType().getPortTypeNameArray();
        for(QName portTypeName:portTypeNames){
            try{
                managedResources = coordinatorContext.getRuleEngine().queryManagedResources("ResolveServiceDependancy", 
                        new String[]{portTypeName.toString()});
            }catch(Throwable e){
                log.warn("Error happend doing a query, retrying",e);
                try{Thread.sleep(2000);}catch (Exception e1) {e1.printStackTrace();};
                managedResources = coordinatorContext.getRuleEngine().queryManagedResources("ResolveServiceDependancy", 
                        new String[]{portTypeName.toString()});
            }
            
            if(managedResources != null){
                ServiceData serviceData = resolveServiceDependancyByTypeResponse.addNewServiceData();
                for(ManagedResource resource:managedResources){
                    serviceData.addNewServiceEndpoint().setStringValue(resource.getManagerAgentURL());
                    serviceData.setServiceName(resource.getQName());
                }
            }
        }
        return responseDocument;
    }
    
    public ExecuteQueryResponseDocument processExecuateQuery(ExecuteQueryDocument document){
        ExecuteQuery executeQuery = document.getExecuteQuery();
        
        ExecuteQueryResponseDocument responseDocument = ExecuteQueryResponseDocument.Factory.newInstance();
        ExecuteQueryResponse executeQueryResponse = responseDocument.addNewExecuteQueryResponse();
        if(executeQuery != null){
            for(Query query:executeQuery.getQueryArray()){
                QueryResults queryResponse = executeQueryResponse.addNewQueryResponse();
                if(query.getQueryName() != null){
                    queryResponse.setQuery(query.getQueryName());
                    List<ManagedResource> managedResources = 
                        coordinatorContext.getRuleEngine().queryManagedResources(query.getQueryName(),query.getParameterArray());
                    for(ManagedResource resource:managedResources){
                        queryResponse.addResource(resource.getQName());    
                    }
                }  else{
                    throw new UnsupportedOperationException("dynamic query installation is not supported yet");
                }             
            }
        }
        return responseDocument;

    }
    
    public void processNotifyEventDocument(NotifyEventDocument eventDocument){
        ActionContext actionContext = coordinatorContext.getRuleEngine().getActionCenter().getActionContext();
        final NotifyEvent notifyEvent = eventDocument.getNotifyEvent();
        actionContext.getEventCorrelator().triggerEvent(new HasthiEvent(notifyEvent.getName(), 
                notifyEvent.getEventDescription(),notifyEvent.getContent()));
    }

    
    public void processUserIntervention(UserInterventionDocument document) throws HasthiException{
        final UserIntervention userIntervention = document.getUserIntervention();
        coordinatorContext.setActiveManagement(userIntervention.getActiveManagement());
        dlogger.info("Set active Management "+userIntervention.getActiveManagement());
        
        if(userIntervention.getServiceStatusChangeArray()!= null)
        for(ServiceStatusChange statusChange:userIntervention.getServiceStatusChangeArray()){
            QName serviceName = statusChange.getServiceName();
            if(statusChange.getRemoveService()){
                UnicastContianier container = coordinatorContext.getMyManagerContext().getUnicastContainer();
                
                UnAssignResourceDocument unAssignResourceDocument = UnAssignResourceDocument.Factory.newInstance();
                unAssignResourceDocument.addNewUnAssignResource().setName(serviceName);
                RemoteManagerContext managerForResource = coordinatorContext.getManagerForResource(serviceName);
                container.sendReceive(new SimpleMessageContext(unAssignResourceDocument,managerForResource.getAddress().getSoapAddress(),
                        HasthiConstants.UNASSIGN_RESOURCE_ACTION));
                coordinatorContext.getResourceModel().removeResource(serviceName);
                coordinatorContext.getRuleEngine().removeObject(serviceName);
            }else{
                EditableManagedResource resource = coordinatorContext.getResourceModel().getResource(serviceName);
                resource.setState(SystemState.valueOf(statusChange.getOperationalStatus()));
                coordinatorContext.getRuleEngine().objectChanged(resource);
            }
        }

    }
    
    
    public void processSuspectFaulty(SuspectFaultyDocument document ) throws HasthiException{
        QName failedServiceName = document.getSuspectFaulty().getServiceName();
        RemoteManagerContext managerForResource = coordinatorContext.getManagerForResource(failedServiceName);
        
        if(managerForResource != null){
            SimpleMessageContext messageContext = new SimpleMessageContext(document, 
                    managerForResource.getAddress().getSoapAddress(),HasthiConstants.SUSPECT_FAULTY_ACTION);
            coordinatorContext.getMyManagerContext().getUnicastContainer().send(messageContext);
        }
    }
    
    public class ResourceUpdater implements XmlBeanUtils.ParameterProcessor{
        private EditableManagedResource resource;
        public ResourceUpdater(EditableManagedResource resource){
            this.resource = resource; 
        }
        public void process(QName name, Object value) throws HasthiException {
             resource.setProperty(name.getLocalPart(), value);
        }
    }
    
}

